package process;

import com.alibaba.fastjson.JSON;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;

/**
 * non keyed process function 使用
 */
public class NonKeyedProcessFunction {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());

        DataStreamSource<String> line = env.socketTextStream("hadoop1", 8888);

        // {"name": "xiaoyu","age": 21,"fv": 99.9}
        // {"name": "xiaoyu","age": 21,"fv": 99.9
        // {"name": "saobing","age": 20,"fv": 89.9,"girls":[{"name": "fengjie"},{"name": "芙蓉姐姐"}]}
        line.process(new ProcessFunction<String, Boy>() {
            @Override
            public void processElement(String value, Context ctx, Collector<Boy> out) throws Exception {
                try {
                    Boy boy = JSON.parseObject(value, Boy.class);
                    if (boy.getGirls() != null) {
                        for (Girl girl : boy.getGirls()) {
                            boy.setGirlName(girl.getName());
                        }
                    } else {
                        out.collect(boy);
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }).print();

        env.execute("");
    }
}
